/*
 * @(#)FlumeAppender.java   1.0  2021年9月7日
 * 
 * Copyright (c)	2014-2020. All Rights Reserved.	GuangZhou hhmk Technology Company LTD.
 */
package com.swift.log.flume.appender;

import java.nio.charset.Charset;
import java.util.Properties;
import java.util.concurrent.locks.ReentrantLock;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.flume.Event;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientConfigurationConstants;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.api.RpcClientFactory.ClientType;
import org.apache.flume.event.EventBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.swift.core.env.EnvLoader;
import com.swift.core.log.appender.SwiftAppender;
import com.swift.core.log.appender.SwiftRemoteAppender;
import com.swift.util.exec.ThreadUtil;
import com.swift.util.type.TypeUtil;

import ch.qos.logback.classic.Level;

/**
 * 添加说明
 * 
 * @author zhengjiajin
 * @version 1.0 2021年9月7日
 */
@Component
public class FlumeAppender extends SwiftAppender {

    private static final Logger log = LoggerFactory.getLogger(FlumeAppender.class);

    private static final int DEFAULT_BATCH_SIZE = 100;
    private static final int DEFAULT_CONNECT_TIMEOUT = 20000;
    private static final int DEFAULT_REQUEST_TIMEOUT = 20000;
    private static final int DEFAULT_IO_WORKERS = Runtime.getRuntime().availableProcessors() * 2;
    private static final int DEFAULT_RECONNECT_TIME = 30000;

    /**
     * All synchronization in this class is done via the lock object.
     */
    protected final ReentrantLock lock = new ReentrantLock(true);

    @Value("${fluem.hosts:}")
    private String hosts;

    @Value("${sysId:}")
    private String sysId;

    private volatile String thisHost;
    private volatile RpcClient rpcClient;
    private Reconnector reconnector;

    /**
     * @see com.swift.core.log.appender.SwiftAppender#doStart()
     */
    @Override
    public boolean doStart() {
        if (TypeUtil.isNull(hosts)) return false;
        if (TypeUtil.isNull(sysId)) return false;
        return true;
    }

    /**
     * @see com.swift.core.log.appender.SwiftAppender#append(java.lang.String)
     */
    @Override
    public void append(Level level, String body) throws Exception {
        if (!doStart()) return;
        Event event = EventBuilder.withBody(body, Charset.defaultCharset());
        event.getHeaders().put("sysId", sysId);
        event.getHeaders().put("env", EnvLoader.getEnv().name());
        event.getHeaders().put("level", level.toString());
        rpcClient.append(event);
    }

    @PostConstruct
    private void start() {
        if (!doStart()) return;
        rpcClientConnect();
        SwiftRemoteAppender.addRemoteAppender(this);
        reconnector = new Reconnector();
        reconnector.start();
    }

    @PreDestroy
    public void stop() {
        lock.lock();
        try {
            if (reconnector != null) {
                reconnector.isTerminated = true;
                reconnector.interrupt();
                try {
                    reconnector.join();
                } catch (InterruptedException ex) {
                }
            }
            if (rpcClient != null) {
                rpcClient.close();
                rpcClient = null;
            }
        } finally {
            lock.unlock();
        }
    }

    private String nextHost() {
        String[] host = hosts.split(",");
        if (host.length == 1) {
            this.thisHost = host[0];
        } else {
            if (TypeUtil.isNull(this.thisHost)) {
                int code = Math.abs(this.sysId.hashCode()) % host.length;
                this.thisHost = host[code];
            } else {
                for (int i = 0; i < host.length; i++) {
                    if (this.thisHost.equals(host[i])) {
                        if (i < host.length - 1) {
                            this.thisHost = host[i + 1];
                        } else {
                            this.thisHost = host[0];
                        }
                        break;
                    }
                }
            }
        }
        log.info("连麦接flume" + this.thisHost);
        return this.thisHost;
    }

    private RpcClient rpcClientConnect(String host) {
        Properties properties = new Properties();
        properties.put(RpcClientConfigurationConstants.CONFIG_CLIENT_TYPE, ClientType.DEFAULT.name());
        properties.put(RpcClientConfigurationConstants.CONFIG_HOSTS, "h1");
        properties.put(RpcClientConfigurationConstants.CONFIG_HOSTS_PREFIX + "h1", host);
        properties.put(RpcClientConfigurationConstants.CONFIG_BATCH_SIZE, DEFAULT_BATCH_SIZE);
        properties.put(RpcClientConfigurationConstants.CONFIG_CONNECT_TIMEOUT, DEFAULT_CONNECT_TIMEOUT);
        properties.put(RpcClientConfigurationConstants.CONFIG_REQUEST_TIMEOUT, DEFAULT_REQUEST_TIMEOUT);
        properties.put(RpcClientConfigurationConstants.MAX_IO_WORKERS, String.valueOf(DEFAULT_IO_WORKERS));
        RpcClient rpcClient = RpcClientFactory.getInstance(properties);
        if (rpcClient == null) return null;
        if (rpcClient.isActive()) {
            return rpcClient;
        } else {
            rpcClient.close();
        }
        return null;
    }

    private void rpcClientConnect() {
        lock.lock();
        try {
            if (rpcClient != null && rpcClient.isActive()) {
                return;
            }
            String[] host = hosts.split(",");
            for (int i = 0; i < host.length; i++) {
                RpcClient createClient = rpcClientConnect(nextHost());
                if (createClient == null) continue;
                rpcClient = createClient;
                return;
            }
        } finally {
            lock.unlock();
        }
    }

    private class Reconnector extends Thread {

        private volatile boolean isTerminated = false;

        public Reconnector() {
            setName("Reconnector");
            setDaemon(true);
        }

        @Override
        public void run() {
            while (!isTerminated) {
                try {
                    rpcClientConnect();
                } catch (Throwable ex) {
                }
                ThreadUtil.sleep(DEFAULT_RECONNECT_TIME);
            }
        }
    }

}
